/*
 * Copyright (c) 2024 Black Sesame Technologies
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <unistd.h>
#include "dal_fastdds/dbh/DalDbhWriter.hpp"
#include "dal_fastdds/utils/log.hpp"

DalDbhWriter::DalDbhWriter()
{
    dbh_send_notify_msg_.global_fd = -1;
    dbh_send_notify_msg_.offset = 0;
    dbh_send_notify_msg_.mmap_size = 0;
    dbh_send_notify_msg_.msg_size = 0;
    dbh_send_notify_msg_.index = 0;
    estimated_size = 0;
    for (auto& ch : dbh_send_notify_msg_.uuid) {
        ch = 0;
    }

    const char* env_var_value = std::getenv("DAL_DBH_DEBUG");
    if (env_var_value != nullptr && std::atoi(env_var_value) == 1) {
        is_debug_print_enabled_ = true;
    }
}

DalDbhWriter::~DalDbhWriter()
{
    // delete notify entities
    fastdds_pub_->delete_datawriter(dbh_send_notify_writer_);
    if (dbh_recv_notify_reader_)
        dbh_recv_notify_subscriber_->delete_datareader(dbh_recv_notify_reader_);
    if (dbh_recv_notify_subscriber_)
        fastdds_participant_->delete_subscriber(dbh_recv_notify_subscriber_);
    fastdds_participant_->delete_topic(dbh_send_notify_topic_);
    fastdds_participant_->delete_topic(dbh_recv_notify_topic_);

    fastdds_pub_->delete_datawriter(dbh_info_writer_);
    fastdds_participant_->delete_topic(dbh_info_topic_);
}

void DalDbhWriter::SetFastddsPublisher(Publisher*& publisher)
{
    fastdds_pub_ = publisher;
}

void DalDbhWriter::SetFastddsDomainParticipant(DomainParticipant*& participant)
{
    fastdds_participant_ = participant;
}

void DalDbhWriter::SetDbhSendNotifyWriterListener(DataWriterListenerImpl* listener)
{
    dbh_send_notify_writer_listener_ = listener;
}

void DalDbhWriter::SetDbhSendNotifyWriterQos(const DataWriterQos& qos)
{
    dbh_send_notify_writer_qos_ = qos;
}

void DalDbhWriter::SetDbhSendNotifyWriterStatusMask(const StatusMask& mask)
{
    dbh_send_notify_writer_status_mask_ = mask;
}

void DalDbhWriter::SetTypeSupport(const void* ts)
{
    if (!ts)
        return;
    msg_ts_callbacks_ = static_cast<const message_type_support_callbacks_t*>(ts);
    if (!msg_ts_) {
        msg_ts_ = new (std::nothrow) MessageTypeSupport_cpp(msg_ts_callbacks_);
    }
}

void DalDbhWriter::SetOriginTopicName(const std::string& origin_topic_name)
{
    origin_topic_name_ = origin_topic_name;
}

bool DalDbhWriter::InitRingBuffer(void* msg)
{
    if (is_first_write_.load()) {
        DmaBufHeapManager* h = DmaBufHeapManager::GetInstance();
        estimated_size = msg_ts_->getEstimatedSerializedSize(msg, msg_ts_callbacks_);
        auto block_size = GetDbhBlockSize(estimated_size);
        if (!ringbuffer_) {
            ringbuffer_ = new BufferDescriptorRingBuffer(RINGBUFFER_LEN);
        }
        if (!ringbuffer_->Init(block_size)) {
            LOG_ERROR_WITH_LOCATION() << "ringbuffer init failed" << std::endl;
            return false;
        }
        is_first_write_.store(false);
    }
    return true;
}

bool DalDbhWriter::CheckResize(void* msg)
{
    std::lock_guard<std::mutex> lock(ringbuffer_->ringbuffer_mutex_);

    auto& buffer_desc_ = ringbuffer_->buffer_[ringbuffer_->GetWriteIndex()];
    auto& block_tmp_ = ringbuffer_->buffer_[ringbuffer_->GetWriteIndex()].dbh_block_;

    // resize block if msg size is larger than block size.
    DmaBufHeapManager* h = DmaBufHeapManager::GetInstance();
    estimated_size = msg_ts_->getEstimatedSerializedSize(msg, msg_ts_callbacks_);
    if (estimated_size > block_tmp_.size) {
        auto block_size = GetDbhBlockSize(estimated_size);
        auto global_fd_removed = block_tmp_.global_fd;
        if (!h->ResizeBlock(block_size, &block_tmp_)) {
            LOG_ERROR_WITH_LOCATION() << "resize block failed" << std::endl;
            return false;
        }

        dal_builtin::msg::DalDmaBufHeapInfo dbh_info_msg_;
        dbh_info_msg_.event = static_cast<uint32_t>(DbhNotifyMsgDef::RESIZE_INFO);
        dbh_info_msg_.data = std::to_string(global_fd_removed);

        // send info msg.
        InstanceHandle_t f_handle;
        dal_fastrtps_shared_cpp::SerializedData sdata;
        sdata.is_cdr_buffer = false;
        sdata.data = static_cast<void*>(&dbh_info_msg_);
        sdata.impl = dbh_info_ts_callbacks_;

        if (DAL_RETCODE_OK != dbh_info_writer_->write(&sdata, f_handle)) {
            LOG_ERROR_WITH_LOCATION() << "dbh info writer write failed. " << std::endl;
            return false;
        }

        if (is_debug_print_enabled_) {
            std::cout << "[dbh info writer ------>] event: RESIZE_INFO, removed_global_fd = "
                      << global_fd_removed << ", new global_fd = " << block_tmp_.global_fd
                      << std::endl;
        }
    }
    return true;
}

DataWriter* DalDbhWriter::GetDbhSendNotifyWriter()
{
    return dbh_send_notify_writer_;
}

bool DalDbhWriter::CreateDbhBuiltinEntities()
{
    return CreateDbhTopic() && CreateDbhSendNotifyWriter() && CreateDbhRecvNotifyReader() &&
           CreateDbhInfoWriter();
}

extern "C" {
extern const dalidl_message_type_support_t*
dalidl_typesupport_fastrtps_cpp__get_message_type_support_handle__dal_builtin__msg__DalDmaBufHeapSendNotifyMsg();
extern const dalidl_message_type_support_t*
dalidl_typesupport_fastrtps_cpp__get_message_type_support_handle__dal_builtin__msg__DalDmaBufHeapRecvNotifyMsg();
extern const dalidl_message_type_support_t*
dalidl_typesupport_fastrtps_cpp__get_message_type_support_handle__dal_builtin__msg__DalDmaBufHeapInfo();
}

bool DalDbhWriter::CreateDbhTopic()
{
    if (origin_topic_name_.length() == 0 || origin_topic_name_ == "") {
        LOG_ERROR_WITH_LOCATION() << "origin topic name is empty" << std::endl;
        return false;
    }
    // !TODO more complex, use hash
    dbh_send_notify_topic_name_ = origin_topic_name_ + "_dbh_send_notify";
    dbh_recv_notify_topic_name_ = origin_topic_name_ + "_dbh_recv_notify";
    dbh_info_topic_name_ = origin_topic_name_ + "_dbh_info";

    if (!fastdds_participant_) {
        return false;
    }

    // We don't load lib but call func directly.
    const dalidl_message_type_support_t* send_notify_ts_handle =
        dalidl_typesupport_fastrtps_cpp__get_message_type_support_handle__dal_builtin__msg__DalDmaBufHeapSendNotifyMsg();
    dbh_send_notify_msg_ts_callbacks_ =
        static_cast<const message_type_support_callbacks_t*>(send_notify_ts_handle->data);
    std::string send_notify_type_name = _create_type_name(dbh_send_notify_msg_ts_callbacks_);
    auto send_notify_tsupport =
        new (std::nothrow) MessageTypeSupport_cpp(dbh_send_notify_msg_ts_callbacks_);
    auto send_notify_ts = new TypeSupport(send_notify_tsupport);
    send_notify_ts->register_type(fastdds_participant_);
    dbh_send_notify_topic_ = fastdds_participant_->create_topic(
        dbh_send_notify_topic_name_, send_notify_type_name, TOPIC_QOS_DEFAULT);
    if (!dbh_send_notify_topic_)
        return false;

    // We don't load lib but call func directly.
    const dalidl_message_type_support_t* recv_notify_ts_handle =
        dalidl_typesupport_fastrtps_cpp__get_message_type_support_handle__dal_builtin__msg__DalDmaBufHeapRecvNotifyMsg();
    dbh_recv_notify_msg_ts_callbacks_ =
        static_cast<const message_type_support_callbacks_t*>(recv_notify_ts_handle->data);
    std::string recv_notify_type_name = _create_type_name(dbh_recv_notify_msg_ts_callbacks_);
    auto recv_notify_tsupport =
        new (std::nothrow) MessageTypeSupport_cpp(dbh_recv_notify_msg_ts_callbacks_);
    auto recv_notify_ts = new TypeSupport(recv_notify_tsupport);
    recv_notify_ts->register_type(fastdds_participant_);
    dbh_recv_notify_topic_ = fastdds_participant_->create_topic(
        dbh_recv_notify_topic_name_, recv_notify_type_name, TOPIC_QOS_DEFAULT);
    if (!dbh_recv_notify_topic_)
        return false;

    // We don't load lib but call func directly.
    const dalidl_message_type_support_t* info_ts_handle =
        dalidl_typesupport_fastrtps_cpp__get_message_type_support_handle__dal_builtin__msg__DalDmaBufHeapInfo();
    dbh_info_ts_callbacks_ =
        static_cast<const message_type_support_callbacks_t*>(info_ts_handle->data);
    std::string dbh_info_type_name = _create_type_name(dbh_info_ts_callbacks_);
    auto dbh_info_tsupport = new (std::nothrow) MessageTypeSupport_cpp(dbh_info_ts_callbacks_);
    auto dbh_info_ts = new TypeSupport(dbh_info_tsupport);
    dbh_info_ts->register_type(fastdds_participant_);
    dbh_info_topic_ = fastdds_participant_->create_topic(
        dbh_info_topic_name_, dbh_info_type_name, TOPIC_QOS_DEFAULT);
    if (!dbh_info_topic_)
        return false;

    return true;
}

bool DalDbhWriter::CreateDbhInfoWriter()
{
    DataWriterQos f_qos;
    f_qos.reliability().kind = RELIABLE_RELIABILITY_QOS;
    f_qos.history().kind = KEEP_ALL_HISTORY_QOS;
    f_qos.history().depth = 5;
    f_qos.durability().kind = VOLATILE_DURABILITY_QOS;
    // default is PREALLOCATED_MEMORY_MODE in fastrtps v2.6.1, but when sending string, we can't
    // judge it's length so must permit realloc.
    f_qos.endpoint().history_memory_policy =
        eprosima::fastrtps::rtps::MemoryManagementPolicy::PREALLOCATED_WITH_REALLOC_MEMORY_MODE;

#ifdef DAL_FASTDDS_DATA_SHARING_DISABLED
    f_qos.data_sharing().off();
#endif

    // !TODO listener
    dbh_info_writer_ =
        fastdds_pub_->create_datawriter(dbh_info_topic_, f_qos, nullptr, StatusMask::none());
    if (dbh_info_writer_) {
        return true;
    }
    return false;
}

bool DalDbhWriter::SendCloseFdsDbhInfo()
{
    if (!ringbuffer_) {
        return true;
    }
    {
        std::lock_guard<std::mutex> lock(ringbuffer_->ringbuffer_mutex_);
        std::string fds;
        if (!Fds2Str(ringbuffer_->GetAllGlobalFds(), fds)) {
            LOG_ERROR_WITH_LOCATION() << "ringbuffer fds2str failed" << std::endl;
        }

        dal_builtin::msg::DalDmaBufHeapInfo dbh_info_msg_;
        dbh_info_msg_.event = static_cast<uint32_t>(DbhNotifyMsgDef::DELETE_FDS);
        dbh_info_msg_.data = fds;

        // send info msg.
        InstanceHandle_t f_handle;
        dal_fastrtps_shared_cpp::SerializedData sdata;

        sdata.is_cdr_buffer = false;
        sdata.data = static_cast<void*>(&dbh_info_msg_);
        sdata.impl = dbh_info_ts_callbacks_;

        if (DAL_RETCODE_OK != dbh_info_writer_->write(&sdata, f_handle)) {
            LOG_ERROR_WITH_LOCATION() << "dbh info writer write failed. " << std::endl;
            return false;
        }

        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        return true;
    }
}

uint32_t DalDbhWriter::GetDbhBlockSize(uint32_t msg_size)
{
    long page_size = sysconf(_SC_PAGE_SIZE);
    if (page_size == -1) {
        return 0;
    }

    uint32_t num_pages = (msg_size + page_size - 1) / page_size;
    return num_pages * page_size;
}